package com.ljm.reactor;

import reactor.core.publisher.Flux;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author 李佳明 https://github.com/pkpk1234
 * @date 2018-04-16
 */
public class FluxWithMySubscriber {
    public static String alphabet(int letterNumber) {
        if (letterNumber < 1 || letterNumber > 26) {
            return null;
        }
        int letterIndexAscii = 'A' + letterNumber - 1;
        return "" + (char) letterIndexAscii;
    }

    public static void main(String[] args) {


        Flux.empty().subscribe(it -> System.out.println("end"));

        //subscirbe with consumer and error handler and completeConsumer
        System.out.println("example for subscribe with consumer , error handler, completeConsumer" +
                "and customerSubscriber");
        Flux<DivideIntegerSupplier> integerFlux = Flux.just(
                new DivideIntegerSupplier(1, 2),
                new DivideIntegerSupplier(8, 2),
                new DivideIntegerSupplier(20, 10),
                new DivideIntegerSupplier(2, 2)
        );
        MySubscriber<DivideIntegerSupplier> integerMySubscriber = new MySubscriber<>();
        integerFlux.subscribe(
                integer ->  integer.get(),
                throwable -> throwable.getMessage(),
                () -> System.out.println("-- No Error and Finished --"),
                integer -> integerMySubscriber.request(5));

        System.out.println("=====================");

        integerFlux.subscribe(integerMySubscriber);
        System.out.println("this is end of main");

        System.out.println("=====================");

        Flux<String> flux = Flux.generate(
                // 初始化一个可变对象作为状态变量。(原生类型及其包装类，以及String等属于不可变类型)
                AtomicLong::new,
                (state, sink) -> {
                    // 改变状态值。
                    long i = state.getAndIncrement();
                    sink.next("3 x " + i + " = " + 3*i);
                    if (i == 10) sink.complete();
                    // 返回 同一个 实例作为新的状态。
                    return state;
                },
                // Comsumer 在序列终止才被调用: 我们会看到最后一个状态值（11）会被这个 Consumer lambda 输出。
                // 可用于清理状态对象
                (state) -> System.out.println("state: " + state));

        Flux.just(-1, 30, 13, 9, 20)
                .handle((i, sink) -> {
                    // 获取数字映射的字母
                    String letter = alphabet(i);
                    if (letter != null)
                        // 如果不是空才会调用 sink.next
                        sink.next(letter);
                }).subscribe(System.out::println);



    }
}
